/**
 *------------------------------------------------------------------------------
 *  @copyright Copyright (c) 2022  <dx_65535@163.com>. 
 * 
 *  @file       : message_poll.cc
 *  @author     : Xiang.D (dx_65535@163.com)
 *  @version    : 1.0
 *  @brief      : Blog: https://linuxtaoist.gitee.io
 *  @date       : 2022-06-04
 * 
 *    Listen for handle events added to epoll and bind the processing interfaces 
 *  of different handles via ep.data.ptr.
 *
 *  Change History:
 *  <Date>     | <Version> | <Author>       | <Description>
 *------------------------------------------------------------------------------
 *  2022/06/04 | 1.0.0.1   | Xiang.D        | Create file
 *------------------------------------------------------------------------------
 * 
 */

#include <stdio.h>
#include <errno.h>
#include <unistd.h>
#include <string.h>
#include <signal.h>
#include <sys/epoll.h>
#include "led.h"
#include "message_epoll.h"
#include "msg_queue.h"

using namespace std;
#define PLOGW(fmt, args...)  printf("%d MsgEpoll W: " fmt, __LINE__, ##args)
#define PLOGE(fmt, args...)  printf("%d MsgEpoll E: " fmt, __LINE__, ##args)

#ifdef TEST_DEBUG
#define PLOGD(fmt, args...)  printf("%d MsgEpoll D: " fmt, __LINE__, ##args)
#else
#define PLOGD(fmt, args...)
#endif

const int EPOLL_FD_NUM = 10;
const int sigArr[] = {SIGINT, SIGTERM, SIGQUIT, SIGPIPE, SIGUSR1};

CMsgEpoll* CMsgEpoll::GetInstance()
{
    static CMsgEpoll Instance(EPOLL_FD_NUM);
    return &Instance;
}

CMsgEpoll::CMsgEpoll(int size)
{
    PLOGD("---- Msg Epoll Start ----\n");

    if (size)
        mFd = epoll_create(size);
    else
        mFd = epoll_create1(0);

    if (mFd < 0) {
        PLOGE("epoll_create failed!\n");
    }

    mQuit = false;
    mMsgQueueMutexLock = PTHREAD_MUTEX_INITIALIZER;
}

CMsgEpoll::~CMsgEpoll()
{

}

static void epoll_thread()
{
    CMsgEpoll::GetInstance()->StartEpoll(true);
}

static void schedule_thread()
{
    CMsgEpoll::GetInstance()->TaskSchedule();
}

void CMsgEpoll::Init()
{
    theMsgQueue.Init();

    // 注册功能响应类
    mCmdTable.insert(std::pair<ECmdType, CBeahaviorBase *>(CMD_LED, CLed::GetInstance()));    // Led
}

void CMsgEpoll::Exit()
{
    PLOGD("Exit from Msg Epoll.\n");

    // 退出线程
    mQuit = true;
    pthread_cancel(mEpollThread.native_handle());
    pthread_cancel(mScheduleThread.native_handle());
    mEpollThread.join();
    mScheduleThread.join();

    if (mFd > 0) {
        close(mFd);
    }
}

void CMsgEpoll::Run()
{
    // 分发消息线程
    mEpollThread = thread(epoll_thread);

    // 读取消息队列消息线程
    mScheduleThread = thread(schedule_thread);
}

/* 生产线程. 写入外部申请的消息至消息队列 */
void CMsgEpoll::StartEpoll(bool run)
{
    struct epoll_event ep[32];
    
    PLOGD("Run epoll thread start (%d).\n", mFd);

    if (run) {
        do {
            // 无事件时, epoll_wait阻塞
            // 增加监听终止信号，避免一直阻塞
            int count = epoll_wait(mFd, ep, sizeof(ep)/sizeof(ep[0]), -1);
            if (count == 0 || count == -1) {
                PLOGE("epoll wait failed. %s\n", strerror(errno));
                continue;
            }

            PLOGD("Get data from epoll ...\n");
            for (int i = 0; i < count; i++) {
                PLOGD("Parse frame...\n");
                SPollEvent *p = (SPollEvent *)ep[i].data.ptr;
                SCmdFrame theDataFrame;

                // 1. 从通道接收数据, 可通过ptr判断是那个通道的数据
                //    将不同通道的数据解析并整理为统一的格式
                if (p && p->pDataParse) {
                    PLOGD("Channel is %d.\n", p->fd);
                    p->pDataParse->ParseFrame(p->fd, &theDataFrame);
                } else {
                    PLOGE("Parse func is NULL!\n");
                }

                // 2. 帧数据分发缓存至消息队列
                //    另起线程读取消息队列，并分发至功能模块
                SMsgEventPacket dataPacket;
                memcpy(&dataPacket, &theDataFrame, sizeof(SCmdFrame));
                dataPacket.handler = p->handler;

                pthread_mutex_lock(&mMsgQueueMutexLock);
                theMsgQueue.Send((const char *)&dataPacket, sizeof(SMsgEventPacket));
                pthread_mutex_unlock(&mMsgQueueMutexLock);
            }
        } while (!mQuit);
    }
    else {
        mQuit = true;
    }
}

void CMsgEpoll::AddPoll(int fd, ETransChannel channel, MsgHanderPtr handler)
{
    struct epoll_event ep;

    PLOGD("Add fd event [%d].\n", fd);

    //EPOLLIN ：表示对应的文件描述符可以读（包括对端SOCKET正常关闭）；
    //EPOLLOUT：表示对应的文件描述符可以写；
    //EPOLLET： 将EPOLL设为边缘触发(Edge Triggered)模式，这是相对于水平触发(Level Triggered)来说的。
    ep.events = EPOLLIN | EPOLLET;

    // 添加监听句柄、事件处理函数
    SPollEvent *pollEvent = new SPollEvent();   // 在DelPoll释放
    pollEvent->fd = fd;
    pollEvent->handler = handler;
    pollEvent->pDataParse = mDataParseFactory.GetChannelParse(channel);
    ep.data.ptr = pollEvent;

    auto it = mEpollTable.find(fd);

    if (it == mEpollTable.end()) {
        mEpollTable.insert(std::pair<int, SPollEvent *>(fd, pollEvent));
    } else {
        PLOGE("fd [%d] invalid.\n", fd);
        return;
    }

    //EPOLL_CTL_ADD：注册新的fd到epfd中；
    //EPOLL_CTL_MOD：修改已经注册的fd的监听事件；
    //EPOLL_CTL_DEL：从epfd中删除一个fd；
    if (epoll_ctl(mFd, EPOLL_CTL_ADD, fd, &ep) != 0) {
        PLOGE("%s\n", strerror(errno));
    }
}

void CMsgEpoll::DelPoll(int fd)
{
    //释放ep.data.ptr
    auto it = mEpollTable.find(fd);
    if (it == mEpollTable.end()) {
        PLOGE("fd [%d] invalid!\n", fd);
        return;
    }

    if (mEpollTable[fd]) {
        delete mEpollTable[fd];
        mEpollTable[fd] = NULL;
        mEpollTable.erase(fd);
    }

    if (epoll_ctl(mFd, EPOLL_CTL_DEL, fd, NULL) != 0) {
        PLOGE("epoll_ctl failed. (%s)\n", strerror(errno));
    }
}

int CMsgEpoll::Invoke(SMsgEventPacket &msgPacket)
{
    int ret = -1;
    ECmdType cmdType = (ECmdType)(msgPacket.frame.cmdAction >> CMD_BYTES);

    PLOGD("Receive cmd type: 0x%x action: 0x%x.\n", cmdType, msgPacket.frame.cmdAction);

    auto it = mCmdTable.find(cmdType);
    if (it == mCmdTable.end()) {
        PLOGE("No this channel: 0x%x\n", msgPacket.frame.cmdAction);
        ret = -1;
    } else {
        ret = mCmdTable[cmdType]->ProcessMsg(msgPacket);
    }

    if (msgPacket.handler) {
        msgPacket.handler(&msgPacket.frame);
    } else {
        PLOGD("handler is NULL!\n");
    }

    return ret;
}

/* 消费线程. 读取消息队列消息，分发至功能模块 */
void CMsgEpoll::TaskSchedule()
{
    SMsgEventPacket theMsgPacket;

    PLOGD("Run shedule thread start.\n");

    do 
    {
        memset(&theMsgPacket, 0x00, sizeof(SMsgEventPacket));
        //pthread_mutex_lock(&mMsgQueueMutexLock);
        if (theMsgQueue.Receive((char *)&theMsgPacket, sizeof(SMsgEventPacket)) >= 0) {
            Invoke(theMsgPacket);
        } 
        sleep(1);
        //pthread_mutex_unlock(&mMsgQueueMutexLock);
    } while(!mQuit);

    PLOGD("Exit thread.\n");
}




